package vms;

import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.tc.vms.MessageWrap;
import com.tc.vms.VMSClient;
import com.tc.vms.VMSEventArgs;
import com.tc.vms.consumer.ConsumeStatus;
import com.tc.vms.encoding.Message;
import com.tc.vms.iface.IPublisher;
import com.tc.vms.iface.ISubscriber;
import com.tc.vms.iface.IVMSCallback;
import com.tc.vms.iface.VmsConsumerContext;
import com.tc.vms.log.VmsClientLogger;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;
import org.slf4j.Logger;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class test {

    @Test
    public void demo(){//日志读取
        //vms.client.log.configFile
        //vms.client.log4j.resource.fileName
        //vms.client.logback.resource.fileName
        System.setProperty("vms.client.log4j.resource.fileName","vmslog/log4j_vms_client.xml");
        System.setProperty("vms.client.logback.resource.fileName","vmslog/logback_vms_client.xml");
        System.setProperty("vms.client.log4j2.resource.fileName","vmslog/log4j2_vms_clients.xml");
        Logger log = VmsClientLogger.getLog();
        log.info("哈哈 我是日志 {}","1");
    }

    @Test
    public void demo1(){//google序列化和反序列化
        Message.VMSMessage.Builder builder=Message.VMSMessage.newBuilder();
        builder.setMessageId("abc");
        builder.setPayload(ByteString.copyFrom("哈哈哈".getBytes()));
        Message.VMSMessage info=builder.build();
        byte[] result=info.toByteArray();
        System.out.println("==========="+result);
        try {
            Message.VMSMessage msg = Message.VMSMessage.parseFrom(result);
            System.out.println(msg.getMessageId());
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    private static final String CONNECT_ADDR = "127.0.0.1:2181";
    //private static final String CONNECT_ADDR = "10.198.195.147:2181,10.198.195.148:2181,10.198.195.149:2181";
    private static final int SESSION_TIMEOUT = 5000;

    @Test
    public void createZkNote() throws Exception { //配置zk mq节点信息(初始)
        //重试策略，初试时间1秒，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        //通过工厂创建Curator
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        //开启连接
        curator.start();
        String defaultPath="/default";
        String defaultData="default" ;

        String channelPath="/default/vms/channel/channel.bfintra.fundout";
        String channelData="vms.inboundName=default_channel_bfintra_fundout\n" +
                "password=osroot\n" +
                "vms.clusterId=/default/vms/cluster/vms.rabbitmq.dev.rabbitmq.default\n" +
                "vms.tps.limit=1\n" +
                "username=root\n" +
                "vhost=default\n" +
                "route.type=direct";
        String rabbitmqPath="/default/vms/cluster/vms.rabbitmq.dev.rabbitmq.default";
        String rabbitmqData="brokerServer=10.198.195.141:5672,10.198.195.142:5672,10.198.195.143:5672\n" +
                "password=osroot\n" +
                "brokerServer.product=rabbitmq\n" +
                "vhost=default\n" +
                "username=root";
        String queuePath="/default/vms/queue/queue.fundout.sendto.merchantnotify";
        String queueData="vms.outboundName=default_queue_fundout_sendto_merchantnotify\n" +
                "password=osroot\n" +
                "vms.clusterId=/default/vms/cluster/vms.rabbitmq.dev.rabbitmq.default\n" +
                "vms.bindings=default_channel_bfintra_fundout.merchantnotify\n" +
                "username=root\n" +
                "vhost=default";
        Map<String,String> map=new HashMap<>();
        map.put(defaultPath,defaultData);
        map.put(channelPath,channelData);
        map.put(rabbitmqPath,rabbitmqData);
        map.put(queuePath,queueData);
        for(Map.Entry<String, String> entry : map.entrySet()){
            Stat stat = curator.checkExists().forPath(entry.getKey());
            if(stat==null){
                curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(entry.getKey(),
                        entry.getValue().getBytes());
            }else{
                curator.setData().forPath(entry.getKey(),entry.getValue().getBytes());
            }
        }
        curator.close();
    }

    /**
     * 发送 单点内部有重试机制
     * @throws Exception
     */
    @Test
    public void send() throws Exception{
        //System.setProperty("VIP_CFGCENTER_ZK_CONNECTION","10.198.195.147:2181,10.198.195.148:2181,10.198.195.149:2181");
        System.setProperty("VIP_CFGCENTER_ZK_CONNECTION","127.0.0.1:2181");

        VMSClient vmsClient = new VMSClient();
        vmsClient.config().domain("osp-fcg.bfintra.com");
        vmsClient.config().appname("osp-fcg.bfintra.com");
        vmsClient.start();

        IPublisher tc = vmsClient.newPublisher("channel.bfintra.fundout");
        tc.start();
        MessageWrap messageWrap = new MessageWrap("abc".getBytes()).addRoutingKey("test-routingkey");
        tc.sendMessage(messageWrap);

        messageWrap = new MessageWrap("cba".getBytes()).addRoutingKey("test-routingkey");
        tc.sendMessage(messageWrap);

    }


    //为每一个队列定义一个多个死信队列，为每个队列定义一个持久队列（用于存储始终消费不掉的）
    //死信队列过期会发往正常消费
    @Test
    public void receive() throws Exception{ //消费
        System.setProperty("VIP_CFGCENTER_ZK_CONNECTION","127.0.0.1:2181");
        VMSClient vmsClient = new VMSClient();
        vmsClient.config().domain("osp-fcg.bfintra.com");
        vmsClient.config().appname("osp-fcg.bfintra.com");
        vmsClient.start();

        ISubscriber iSubscriber = vmsClient.newSubscriber("queue.fundout.sendto.merchantnotify", "1212");
        iSubscriber.qosOption()
                .autoCommit(false)                  //用于选择自动提交或手动提交
                .prefetchCount(64);                  // 用于rabbitmq选项，用于一次request拉取的最大消息条目
        iSubscriber.registerListener(new IVMSCallback() {
            @Override
            public ConsumeStatus consumeMessage(VmsConsumerContext context, VMSEventArgs args) {
                MessageWrap msg = args.getMessage();
                String data = new String(msg.getPayload());
                log.info("消费了:{}",data);
                if(data.equals("cba")){//重试
                    return ConsumeStatus.RECONSUME_LATER;
                }
                return ConsumeStatus.CONSUME_SUCCESS;
            }
        });
        iSubscriber.start();
        Thread.sleep(50000);
    }


    @Test
    public void createZkNote2() throws Exception { //配置zk mq节点信息
        //重试策略，初试时间1秒，重试10次
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        //通过工厂创建Curator
        CuratorFramework curator = CuratorFrameworkFactory.builder().connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT).retryPolicy(policy).build();
        //开启连接
        curator.start();
        String defaultPath="/default";
        String defaultData="default" ;

        String channelPath="/default/vms/channel/channel.bfintra.fundout";
        String channelData="vms.inboundName=test-exchange\n" + //vms.inboundName为exchange
                "password=123456\n" +
                "vms.clusterId=/default/vms/cluster/vms.rabbitmq.dev.rabbitmq.default\n" +
                "vms.tps.limit=1\n" +
                "username=tc\n" +
                "vhost=/\n" +
                "route.type=direct";

        String rabbitmqPath="/default/vms/cluster/vms.rabbitmq.dev.rabbitmq.default";
        String rabbitmqData="brokerServer=10.198.197.127:5672\n" +
                "password=123456\n" +
                "brokerServer.product=rabbitmq\n" +
                "vhost=/\n" +
                "username=tc";
        String queuePath="/default/vms/queue/queue.fundout.sendto.merchantnotify";
        String queueData="vms.outboundName=tc-queues\n" + //vms.outboundName为queues
                "password=123456\n" +
                "vms.clusterId=/default/vms/cluster/vms.rabbitmq.dev.rabbitmq.default\n" +
                "vms.bindings=\n" + //rabbit mq 没用到
                "username=tc\n" +
                "vhost=/";
        Map<String,String> map=new HashMap<>();
        map.put(defaultPath,defaultData);
        map.put(channelPath,channelData);
        map.put(rabbitmqPath,rabbitmqData);
        map.put(queuePath,queueData);
        for(Map.Entry<String, String> entry : map.entrySet()){
            Stat stat = curator.checkExists().forPath(entry.getKey());
            if(stat==null){
                curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(entry.getKey(),
                        entry.getValue().getBytes());
            }else{
                curator.setData().forPath(entry.getKey(),entry.getValue().getBytes());
            }
        }
        curator.close();
    }


}
